package com.infoauto.opc.gateway.messages.model;


import com.infoauto.opc.gateway.messages.constants.MqttConstants;
import com.infoauto.opc.gateway.messages.handler.MqttListener;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

//MQTT协议订阅端
public class MqttConsumer extends MqttAccess {

    //用于记录已订阅的主题过滤器，用于重置订阅
    private List<String> topicFilters = new ArrayList<>();

    private Map<String, Integer> qosMap = new HashMap<>();

    public MqttConsumer(MqttPahoClientFactory clientFactory, String clientId, MqttListener mqttListener) {
        super(clientFactory, clientId, mqttListener);
    }

    /**
     * 订阅主题
     * @param topicFilter
     * @throws MqttException
     */
    public synchronized void subscribe(String topicFilter) throws MqttException {
        subscribe(topicFilter, MqttConstants.DEFAULT_QOS);
    }

    /**
     * 订阅主题
     * @param topicFilter
     * @param qos
     * @throws MqttException
     */
    private synchronized void subscribe(String topicFilter, int qos) throws MqttException {
        if (!topicFilters.contains(topicFilter)) {
            topicFilters.add(topicFilter);
            qosMap.put(topicFilter, qos);
            mqttClient.subscribe(topicFilter, qos);
        }
    }

    /**
     * 取消订阅
     * @param topicFilter
     * @throws MqttException
     */
    public synchronized void unsubscribe(String topicFilter) throws MqttException {
        if (topicFilters.contains(topicFilter)) {
            topicFilters.remove(topicFilter);
            qosMap.remove(topicFilter);
            mqttClient.unsubscribe(topicFilter);
        }
    }

    /**
     * 重置订阅
     * @throws MqttException
     */
    public synchronized void reset() throws MqttException {
        for (String topicFilter : topicFilters) {
            int qos = qosMap.get(topicFilter);
            unsubscribe(topicFilter);
            subscribe(topicFilter, qos);
        }
    }

    /**
     * 清除订阅
     * @throws MqttException
     */
    public synchronized void clear() throws MqttException {
        if (topicFilters.size() > 0) {
            topicFilters.clear();
            qosMap.clear();
            mqttClient.unsubscribe(topicFilters.toArray(new String[topicFilters.size()]));
        }
    }
}
